/*
 * Copyright (c) 2016, www.zm0618.com All Rights Reserved..
 */

package com.ibetter.mr.grade.medic.tag;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * Created by zhaojun on 2016/12/19.
 */
public class GradeUserTagScoreReducer extends Reducer<LongWritable,Text,LongWritable,Text>{

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //key文章ID

        List<Long> catIdList=new ArrayList<Long>();
        Iterator<Text> iter = values.iterator();
        List<GradeUserTagScoreReducer.UserScore> scoreList=new ArrayList<GradeUserTagScoreReducer.UserScore>();
        while (iter.hasNext()){
            Text nextValue = iter.next();
            String valueStr=nextValue.toString();
            if (StringUtils.isNotBlank(StringUtils.trim(valueStr))){
                String[] valuePart = StringUtils.split(valueStr, ":");
                if (StringUtils.equals(valuePart[0],"r")){//分类行  分类ID
                    long catId = Long.parseLong(StringUtils.trim(valuePart[1]));
                    if (catIdList.indexOf(catId)<0){
                        catIdList.add(catId);
                    }
                }

                if (StringUtils.equals(valuePart[0],"s")){//文章行  用户ID  分数
                    long userId = Long.parseLong(StringUtils.trim(valuePart[1]));
                    float scoreFloat = Float.parseFloat(StringUtils.trim(valuePart[2]));
                    GradeUserTagScoreReducer.UserScore userScore=new GradeUserTagScoreReducer.UserScore();
                    userScore.score=scoreFloat;
                    userScore.userId=userId;
                    scoreList.add(userScore);
                }
            }
        }

        //笛卡尔乘积  有数据倾斜
        for (Long catId:catIdList) {
            for (GradeUserTagScoreReducer.UserScore userScore:scoreList){
                Text tx=new Text(StringUtils.join(new String[]{catId.toString(),String.valueOf(userScore.score)},"\t"));
                context.write(new LongWritable(userScore.userId),tx);
            }
        }
    }

    class UserScore implements Serializable {
        protected long userId;
        protected float score;
    }
}
